from typing import Tuple, Union

import numpy as np
import pandas as pd

# import talib
from vnpy.bq.indicator import pd_talib as talib
from vnpy.trader.object import BarData


class PdManager(object):
    """
    For:
    1. time series container of bar data
    2. calculating technical indicator value
    """

    def __init__(self, size: int = 100):
        """Constructor"""
        self.count: int = 0
        self.size: int = size
        self.inited: bool = False

        self.open_series: pd.Series = pd.Series(np.zeros(size))
        self.high_series: pd.Series = pd.Series(np.zeros(size))
        self.low_series: pd.Series = pd.Series(np.zeros(size))
        self.close_series: pd.Series = pd.Series(np.zeros(size))
        self.volume_series: pd.Series = pd.Series(np.zeros(size))
        self.open_interest_series: pd.Series = pd.Series(np.zeros(size))
        self.dt_series: pd.Series = pd.Series(np.zeros(size))

    def update_bar(self, bar: BarData) -> None:
        """
        Update new bar data into array manager.
        """
        self.count += 1
        if not self.inited and self.count >= self.size:
            self.inited = True

        self.open_series[:-1] = self.open_series[1:]
        self.high_series[:-1] = self.high_series[1:]
        self.low_series[:-1] = self.low_series[1:]
        self.close_series[:-1] = self.close_series[1:]
        self.volume_series[:-1] = self.volume_series[1:]
        self.open_interest_series[:-1] = self.open_interest_series[1:]
        self.dt_series[:-1] = self.dt_series[1:]

        self.open_series[-1] = bar.open_price
        self.high_series[-1] = bar.high_price
        self.low_series[-1] = bar.low_price
        self.close_series[-1] = bar.close_price
        self.volume_series[-1] = bar.volume
        self.open_interest_series[-1] = bar.open_interest
        self.dt_series[-1] = bar.datetime

    @property
    def open(self) -> pd.Series:
        """
        Get open price time series.
        """
        return self.open_series

    @property
    def high(self) -> pd.Series:
        """
        Get high price time series.
        """
        return self.high_series

    @property
    def low(self) -> pd.Series:
        """
        Get low price time series.
        """
        return self.low_series

    @property
    def close(self) -> pd.Series:
        """
        Get close price time series.
        """
        return self.close_series

    @property
    def volume(self) -> pd.Series:
        """
        Get trading volume time series.
        """
        return self.volume_series

    @property
    def datetime(self) -> pd.Series:
        """
        get trading datetime series.
        """
        return self.dt_series

    @property
    def open_interest(self) -> pd.Series:
        """
        Get trading volume time series.
        """
        return self.open_interest_series

    def sma(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        Simple moving average.
        """
        result = talib.SMA(self.close, n)
        if is_series:
            return result
        return result[-1]

    def ema(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        Exponential moving average.
        """
        result = talib.EMA(self.close, n)
        if is_series:
            return result
        return result[-1]

    def kama(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        KAMA.
        """
        result = talib.KAMA(self.close, n)
        if is_series:
            return result
        return result[-1]

    def wma(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        WMA.
        """
        result = talib.WMA(self.close, n)
        if is_series:
            return result
        return result[-1]

    def apo(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        APO.
        """
        result = talib.APO(self.close, n)
        if is_series:
            return result
        return result[-1]

    def cmo(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        CMO.
        """
        result = talib.CMO(self.close, n)
        if is_series:
            return result
        return result[-1]

    def mom(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        MOM.
        """
        result = talib.MOM(self.close, n)
        if is_series:
            return result
        return result[-1]

    def ppo(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        PPO.
        """
        result = talib.PPO(self.close, n)
        if is_series:
            return result
        return result[-1]

    def roc(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        ROC.
        """
        result = talib.ROC(self.close, n)
        if is_series:
            return result
        return result[-1]

    def rocr(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        ROCR.
        """
        result = talib.ROCR(self.close, n)
        if is_series:
            return result
        return result[-1]

    def rocp(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        ROCP.
        """
        result = talib.ROCP(self.close, n)
        if is_series:
            return result
        return result[-1]

    def rocr_100(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        ROCR100.
        """
        result = talib.ROCR100(self.close, n)
        if is_series:
            return result
        return result[-1]

    def trix(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        TRIX.
        """
        result = talib.TRIX(self.close, n)
        if is_series:
            return result
        return result[-1]

    def std(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        Standard deviation.
        """
        result = talib.STDDEV(self.close, n)
        if is_series:
            return result
        return result[-1]

    def obv(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        OBV.
        """
        result = talib.OBV(self.close, self.volume)
        if is_series:
            return result
        return result[-1]

    def cci(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        Commodity Channel Index (CCI).
        """
        result = talib.CCI(self.high, self.low, self.close, n)
        if is_series:
            return result
        return result[-1]

    def atr(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        Average True Range (ATR).
        """
        result = talib.ATR(self.high, self.low, self.close, n)
        if is_series:
            return result
        return result[-1]

    def natr(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        NATR.
        """
        result = talib.NATR(self.high, self.low, self.close, n)
        if is_series:
            return result
        return result[-1]

    def rsi(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        Relative Strenght Index (RSI).
        """
        result = talib.RSI(self.close, n)
        if is_series:
            return result
        return result[-1]

    def macd(
            self,
            fast_period: int,
            slow_period: int,
            signal_period: int,
            is_series: bool = False
    ) -> Union[
        Tuple[pd.Series, pd.Series, pd.Series],
        Tuple[float, float, float]
    ]:
        """
        MACD.
        """
        macd, signal, hist = talib.MACD(
            self.close, fast_period, slow_period, signal_period
        )
        if is_series:
            return macd, signal, hist
        return macd[-1], signal[-1], hist[-1]

    def adx(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        ADX.
        """
        result = talib.ADX(self.high, self.low, self.close, n)
        if is_series:
            return result
        return result[-1]

    def adxr(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        ADXR.
        """
        result = talib.ADXR(self.high, self.low, self.close, n)
        if is_series:
            return result
        return result[-1]

    def dx(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        DX.
        """
        result = talib.DX(self.high, self.low, self.close, n)
        if is_series:
            return result
        return result[-1]

    def minus_di(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        MINUS_DI.
        """
        result = talib.MINUS_DI(self.high, self.low, self.close, n)
        if is_series:
            return result
        return result[-1]

    def plus_di(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        PLUS_DI.
        """
        result = talib.PLUS_DI(self.high, self.low, self.close, n)
        if is_series:
            return result
        return result[-1]

    def willr(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        WILLR.
        """
        result = talib.WILLR(self.high, self.low, self.close, n)
        if is_series:
            return result
        return result[-1]

    def ultosc(self, is_series: bool = False) -> Union[float, pd.Series]:
        """
        Ultimate Oscillator.
        """
        result = talib.ULTOSC(self.high, self.low, self.close)
        if is_series:
            return result
        return result[-1]

    def trange(self, is_series: bool = False) -> Union[float, pd.Series]:
        """
        TRANGE.
        """
        result = talib.TRANGE(self.high, self.low, self.close)
        if is_series:
            return result
        return result[-1]

    def boll(
            self,
            n: int,
            dev: float,
            array: bool = False
    ) -> Union[
        Tuple[pd.Series, pd.Series],
        Tuple[float, float]
    ]:
        """
        Bollinger Channel.
        """
        mid = self.sma(n, array)
        std = self.std(n, array)

        up = mid + std * dev
        down = mid - std * dev

        return up, down

    def keltner(
            self,
            n: int,
            dev: float,
            array: bool = False
    ) -> Union[
        Tuple[pd.Series, pd.Series],
        Tuple[float, float]
    ]:
        """
        Keltner Channel.
        """
        mid = self.sma(n, array)
        atr = self.atr(n, array)

        up = mid + atr * dev
        down = mid - atr * dev

        return up, down

    def donchian(
            self, n: int, is_series: bool = False
    ) -> Union[
        Tuple[pd.Series, pd.Series],
        Tuple[float, float]
    ]:
        """
        Donchian Channel.
        """
        up = talib.MAX(self.high, n)
        down = talib.MIN(self.low, n)

        if is_series:
            return up, down
        return up[-1], down[-1]

    def aroon(
            self,
            n: int,
            is_series: bool = False
    ) -> Union[
        Tuple[pd.Series, pd.Series],
        Tuple[float, float]
    ]:
        """
        Aroon indicator.
        """
        aroon_up, aroon_down = talib.AROON(self.high, self.low, n)

        if is_series:
            return aroon_up, aroon_down
        return aroon_up[-1], aroon_down[-1]

    def aroonosc(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        Aroon Oscillator.
        """
        result = talib.AROONOSC(self.high, self.low, n)

        if is_series:
            return result
        return result[-1]

    def minus_dm(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        MINUS_DM.
        """
        result = talib.MINUS_DM(self.high, self.low, n)

        if is_series:
            return result
        return result[-1]

    def plus_dm(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        PLUS_DM.
        """
        result = talib.PLUS_DM(self.high, self.low, n)

        if is_series:
            return result
        return result[-1]

    def mfi(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        Money Flow Index.
        """
        result = talib.MFI(self.high, self.low, self.close, self.volume, n)
        if is_series:
            return result
        return result[-1]

    def ad(self, is_series: bool = False) -> Union[float, pd.Series]:
        """
        AD.
        """
        result = talib.AD(self.high, self.low, self.close, self.volume)
        if is_series:
            return result
        return result[-1]

    def adosc(self, n: int, is_series: bool = False) -> Union[float, pd.Series]:
        """
        ADOSC.
        """
        result = talib.ADOSC(self.high, self.low, self.close, self.volume, n)
        if is_series:
            return result
        return result[-1]

    def bop(self, is_series: bool = False) -> Union[float, pd.Series]:
        """
        BOP.
        """
        result = talib.BOP(self.open, self.high, self.low, self.close)

        if is_series:
            return result
        return result[-1]

    def zig(self, is_series: bool = False) -> Union[float, pd.Series]:
        """
        ZIGZAG.
        """

    def icu(self,periods:int=11,price_sign='close'):
        """

        """
        price=self.close
        lp=price[-1]
        slope=price/price.shift(periods=-periods,fill_value=lp)
        icu=slope.rolling(periods,min_periods=1).median()
        return icu
